/**
 * Copyright (C) 2010, 2011 Neofonie GmbH
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

/**
 *
 */
package eu.dicodeproject.analysis.restapi;

import eu.dicodeproject.analysis.restapi.oozie.TwitterVectorExporter;
import twitter4j.internal.org.json.JSONException;
import twitter4j.internal.org.json.JSONObject;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.SortedMap;
import java.util.SortedSet;
import java.util.TreeMap;

import javax.annotation.Resource;
import javax.servlet.http.HttpServletResponse;

import org.apache.commons.compress.archivers.ArchiveException;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.WritableByteArrayComparable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.oozie.client.OozieClientException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;

/**
 *
 */
@Controller
@RequestMapping("/twitter/")
public class TwitterController {

  private String hbaseTablename;
  private byte[] hbaseColumn;
  private HTablePoolHandler hTablePoolHandler;
  private byte[] hbaseColumnFamily;
  private TwitterVectorExporter exportService;
  private Properties exportProperties;
  private String twitterstreamTablename;

  // TODO: define schema for results:
  // - use a separate column for each meta-data type?
  // - use versions with fixed date instead of timestamps in key?
  // - or: use only one column for all results
  // - where goes the "query"?

  private static final String HASHTAGS_COLUMN = "tag";

  private static final int JSON_INDENT_FACTOR = 2;

  /** Logger for this class. */
  private static final Logger log = LoggerFactory.getLogger(TwitterController.class);

  /**
   * Returns hashtags for a certain topic and day if available
   * This method was implemented to demonstrate the workflow and will be
   * replaced by the generic method "metadata"
   * @param topic
   * @param day
   * @param response
   * @throws IOException
   */
  @RequestMapping(value = "hashtag/{topic}_{day}", method = RequestMethod.GET)
  public void hashtag(@PathVariable String topic, @PathVariable String day, HttpServletResponse response)
      throws IOException {

    response.setContentType("application/json; charset=UTF-8");

    HTableInterface hTable = this.hTablePoolHandler.getHTable(this.hbaseTablename);

    Get get = new Get((topic + "_" + day).getBytes());
    get.addColumn(this.hbaseColumnFamily, "HASHTAGS_COLUMN".getBytes());
    Result result = hTable.get(get);
    final byte[] value = result.value();
    response.setContentLength(value.length);
    this.hTablePoolHandler.putHTable(hTable);
    OutputStream out = response.getOutputStream();
    out.write(value);
    out.flush();
    out.close();
  }

  /**
   * Generic method which returns Twitter meta-data from HBase
   * http://localhost:9889/restapi/twitter/metadata/lang.json?day=20110824
   * @param type
   * @param day
   * @param response
   * @throws IOException
   */
  @RequestMapping(value = "metadata/{type}.json", method = RequestMethod.GET)
  public void metadata(@PathVariable String type,  @RequestParam String day, @RequestParam String callback, HttpServletResponse response)
      throws IOException {

    response.setContentType("text/plain; charset=UTF-8");

    HTableInterface hTable = this.hTablePoolHandler.getHTable(this.hbaseTablename);

    Get get = new Get((type + "_" + day).getBytes());
    get.addColumn(this.hbaseColumnFamily, this.hbaseColumn);
    Result result = hTable.get(get);

    OutputStream out = response.getOutputStream();

    String jsonPrefix = "(" + callback + "(";
    String jsonPostfix = "))";
    String emptyJson = jsonPrefix + "{}" + jsonPostfix;

    if (!result.isEmpty()) {
      final byte[] value = result.value();

      this.hTablePoolHandler.putHTable(hTable);
      out.write(jsonPrefix.getBytes());
      out.write(value);
      out.write(jsonPostfix.getBytes());
    }
    else {
      out.write(emptyJson.getBytes());
    }
    out.flush();
    out.close();
  }

  @RequestMapping(value = "vectors", method = RequestMethod.GET)
  public void vectors(@RequestParam(required = false) String start, @RequestParam(required = false) String stop,
      @RequestParam(required = false) String language, @RequestParam(required = false) String regex,
      @RequestParam String callback, HttpServletResponse response)
      throws IOException, JSONException {

    // parameters
    //TODO hard-coded
    byte[] dataFamily = Bytes.toBytes("d");
    byte[] vectorFamily = Bytes.toBytes("v");
    byte[] textQualifier = Bytes.toBytes("text");
    byte[] langQualifier = Bytes.toBytes("lang");

    // set HBase Scan
    HTableInterface hTable = this.hTablePoolHandler.getHTable(this.hbaseTablename);
    Scan scan = new Scan();
    scan.addFamily(dataFamily); // not part of result but of filters
    scan.addFamily(vectorFamily);

    // collect parameters for result
    Map<String,String> params = new TreeMap<String, String>();

    // set time range
    if (start != null && stop != null) {
      scan.setTimeRange(Long.parseLong(start), Long.parseLong(stop));  //TODO assumes millisecond longs as input
      params.put("start", start);
      params.put("stop", start);
    }
    else if (start == null && stop != null) {
      scan.setTimeRange(0, Long.parseLong(stop));  //TODO assumes millisecond longs as input
      params.put("stop", start);
    }
    else if (start != null && stop == null) {
      scan.setTimeRange(Long.parseLong(start), Long.MAX_VALUE);  //TODO assumes millisecond longs as input
      params.put("start", start);
    }
    //scan.setStartRow(Bytes.toBytes(start));  //assumes that date is in row key
    //scan.setStoptRow(Bytes.toBytes(stop));  //assumes that date is in row key

    // filter based on language and regex
    FilterList filters = new FilterList(FilterList.Operator.MUST_PASS_ALL);
    if (language != null) {
      byte[] l = Bytes.toBytes(language);
      Filter langFilter = new SingleColumnValueFilter(dataFamily, langQualifier, CompareFilter.CompareOp.EQUAL, l);
      filters.addFilter(langFilter);
      params.put("language", language);
    }
    if (regex != null) {
      WritableByteArrayComparable r = new RegexStringComparator(regex);
      Filter regexFilter = new SingleColumnValueFilter(dataFamily, textQualifier, CompareFilter.CompareOp.EQUAL, r);
      filters.addFilter(regexFilter);
      params.put("regex", regex);
    }
    scan.setFilter(filters);

    // read results from scan into a list of maps
    List<Map<String,Object>> vectorsList = new LinkedList<Map<String,Object>>();
    for (Result result : hTable.getScanner(scan)) {
      Map<String,Object> tweet = new TreeMap<String,Object>();
      String id = Bytes.toString(result.getRow());
      //id = id.substring(id.indexOf('_')+1);  //adjust if row key contains date
      tweet.put("id", id);

      Map<String,Integer> vec = new TreeMap<String,Integer>();
      for (Map.Entry<byte[], byte[]> entry : result.getFamilyMap(vectorFamily).entrySet()) {
        String term = Bytes.toString(entry.getKey());
        int count = Bytes.toInt(entry.getValue());
        vec.put(term, count);
      }
      tweet.put("v", vec);
      vectorsList.add(tweet);
    }

    // put result list and parameters map into a JSON
    JSONObject json = new JSONObject();
    json.put("parameters", params);
    json.put("vectors", vectorsList);

    // write out the result JSON
    response.setContentType("text/plain; charset=UTF-8");
    OutputStream out = response.getOutputStream();
    String jsonPrefix = "(" + callback + "(";
    String jsonPostfix = "))";
    out.write(Bytes.toBytes(jsonPrefix + json.toString(JSON_INDENT_FACTOR) + jsonPostfix));
    out.flush();
    out.close();
  }


  /**
   *
   * @param topic
   * @param minSupport
   * @param response
   * @throws IOException
   */
  @Deprecated
  @RequestMapping(value = "export/{topic}_{minSupport}", method = RequestMethod.GET)
  public void export(@PathVariable String topic, @PathVariable String minSupport,
      HttpServletResponse response) {

    InputStream in = null;
    OutputStream out = null;

    try {

      if (!this.validateParams(topic, minSupport)) {
        // response.setStatus(404); // No such resource
        response.sendError(404);
      } else {
        File zipFile = null;

        zipFile = exportService.getTwitterVectors(this.exportProperties);

        if (zipFile != null && zipFile.isFile()) {
          response.setContentType("application/zip; charset=UTF-8");
          response.setHeader("Content-Length", String.valueOf(zipFile.length()));
          response.setHeader("Content-Disposition", "attachment; filename=\"" + zipFile.getName() + "\"");

          byte buff[] = new byte[8192];
          in = new FileInputStream(zipFile);
          out = response.getOutputStream();

          int i = 0;
          while ((i = in.read(buff)) > 0) {
            out.write(buff, 0, i);
            out.flush();
          }
          in.close();
          out.close();
        } else {
          // response.setStatus(403); // Error, no zip file
          response.sendError(403);
        }
      }
    } catch (OozieClientException e) {
      log.error("Error while exporting Tweets: " + e.getMessage());
    } catch (InterruptedException e) {
      log.error("Error while exporting Tweets: " + e.getMessage());
    } catch (ArchiveException e) {
      log.error("Error while writing Tweet archive file: " + e.getMessage());
    } catch (IOException e) {
      log.error("Error in Tweet Export: " + e.getMessage());
    } finally {
      if (out != null) {
        try {
          out.close();
        } catch (IOException e) {
          log.error("Error in Tweet Export: " + e.getMessage());
        }
      }
      if (in != null) {
        try {
          in.close();
        } catch (IOException e) {
          log.error("Error in Tweet Export: " + e.getMessage());
        }
      }
    }
  }

  /**
   * Check parameter AND add default parameters
   *
   * @param topic
   * @param minSupport
   * @return
   */
  private boolean validateParams(String topic, String minSupport) {

    if (topic == null) {
      return false;
    }
    try {
      // check if minSupport is int
      Integer.parseInt(minSupport);
      this.exportProperties = new Properties();
      this.exportProperties.put("hbaseTable", twitterstreamTablename);
      this.exportProperties.put("topic", topic);
      this.exportProperties.put("minSupport", minSupport);
    } catch (NumberFormatException e) {
      return false;
    }
    return true;
  }

  /**
   * Takes a date type as parameter.
   * Returns a JSON that consists of {"result": MAP}, with MAP containing dates
   * (years, months and days) with their frequencies with respect to Tweet creation dates.
   * @param dateType one of {"year", "yearMonth", "yearMonthDay", "hour"}
   *
   * Returns a JSON that consists of {"categoriesLabel": String, "categories": Array, "chartData": Map},
   * with Map containing "creationDates" as key and a map with (date, frequency) as value.
   * Array contains all date values in the required range, and categoriesLabel is the label for these values e.g.
   * {
   *   "categoriesLabel": "yearMonth",
   *   "categories": ["2011-01", "2011-02"],
   *   "chartData": {
   *     "creationDates": {
   *       "2011-01": 54,
   *       "2011-02": 21
   *     }
   *   }
   * }
   * @param callback arbitrary callback parameter
   * @param start earliest date in the result
   * @param stop EXCLUSIVE latest date in the result
   * @param addZeros flag to specify if missing values should be represented by zero entries
   * @param response to write the answer to
   * @throws IOException inherited from HBase Scan and HttpServletResponse.write
   * @throws JSONException inherited from JSONObject constructor
   */
  @RequestMapping(value = "creationDates/{dateType}", method = RequestMethod.GET)
  public void creationDates(@PathVariable String dateType, @RequestParam String callback,
      @RequestParam(required = false) String start, @RequestParam(required = false) String stop,
      @RequestParam(required = false, defaultValue = "false") boolean addZeros,
      HttpServletResponse response)
      throws IOException, JSONException {

    // parameters
    String table = "thtweets-creationDates";  //TODO hard-coded
    String fam = "d";
    String qual = dateType;  // the dataType is the column qualifier (e.g. yearMonth or yearMonthDay)

    String jsonPrefix = "(" + callback + "(";
    String jsonPostfix = "))";

    // set HBase Scan
    HTableInterface hTable = this.hTablePoolHandler.getHTable(table);
    Scan scan = new Scan();
    scan.addColumn(Bytes.toBytes(fam), Bytes.toBytes(qual));

    if (start != null) {
      scan.setStartRow(Bytes.toBytes(start));
    }
    if (stop != null) {
      scan.setStopRow(Bytes.toBytes(stop));
    }

    // read results from scan into a map
    SortedMap<String,Integer> dateMap = new TreeMap<String,Integer>();
    for (Result result : hTable.getScanner(scan)) {
      String date = Bytes.toString(result.getRow());
      int count = Integer.valueOf(Bytes.toString(result.getValue(Bytes.toBytes(fam), Bytes.toBytes(qual))));
      dateMap.put(date, count);
    }

    // get all dates in the required range
    if (start == null) {
      start = dateMap.firstKey();
    }
    SortedSet<String> categories = TwitterControllerUtil.getAllDates(dateType, start, stop);

    // add zeros if required
    if (addZeros) {
      TwitterControllerUtil.addZeros(categories, dateMap);
    }

    // put map into a JSON
    JSONObject json = new JSONObject();
    json.put("categoriesLabel", dateType);
    json.put("categories", categories);
    Map<String,Map<String,Integer>> chartData = new HashMap<String,Map<String,Integer>>();
    chartData.put("creationDates", dateMap);
    json.put("chartData", chartData);

    // write out the result JSON
    response.setContentType("text/plain; charset=UTF-8");
    OutputStream out = response.getOutputStream();
    out.write(Bytes.toBytes(jsonPrefix + json.toString(JSON_INDENT_FACTOR) + jsonPostfix));
    out.flush();
    out.close();
  }

  /**
   * Takes terms as input.
   * Returns a JSON that consists of {"categoriesLabel": String, "categories": Array, "chartData": Map},
   * with Map containing terms as keys and maps with (date, frequency) as values.
   * Array contains all date values in the required range, and categoriesLabel is the label for these values e.g.
   * {
   *   "categoriesLabel": "date",
   *   "categories": ["2011-01-01", "2011-01-02"],
   *   "chartData": {
   *     "google": {
   *       "2011-01-01": 54,
   *       "2011-01-02": 21
   *     }
   *     "microsoft": {
   *       "2011-01-01": 43
   *     }
   *   }
   * }
   * @param terms list of terms separated by commas
   * @param callback arbitrary callback parameter
   * @param start earliest date in the result
   * @param stop EXCLUSIVE latest date in the result
   * @param addZeros flag to specify if missing values should be represented by zero entries
   * @param response to write the answer to
   * @throws IOException inherited from HBase Scan and HttpServletResponse.write
   * @throws JSONException inherited JSONObject
   */
  @RequestMapping(value = "termFreqs", method = RequestMethod.GET)
  public void termFreqs(@RequestParam String terms, @RequestParam String callback,
      @RequestParam(required = false) String start, @RequestParam(required = false) String stop,
      @RequestParam(required = false, defaultValue = "false") boolean addZeros,
      HttpServletResponse response)
      throws IOException, JSONException {

    // parameters
    String table = "thtweets-wordCounts";  //TODO hard-coded
    byte[] columnFamily = Bytes.toBytes("d");

    String jsonPrefix = "(" + callback + "(";
    String jsonPostfix = "))";

    String[] termsArray = terms.split(",");

    // set HBase Scan
    HTableInterface hTable = this.hTablePoolHandler.getHTable(table);
    Scan scan = new Scan();
    for (String t : termsArray) {
      scan.addColumn(columnFamily, Bytes.toBytes(t.trim()));
    }
    if (start != null) {
      scan.setStartRow(Bytes.toBytes(start));
    }
    if (stop != null) {
      scan.setStopRow(Bytes.toBytes(stop));
    }

    // read results from scan into a sorted set and a map
    SortedMap<String,SortedMap<String,Integer>> chartData = new TreeMap<String,SortedMap<String,Integer>>();
    for (Result result : hTable.getScanner(scan)) {
      for (Map.Entry<byte[], byte[]> entry : result.getFamilyMap(columnFamily).entrySet()) {
        String term = Bytes.toString(entry.getKey());
        SortedMap<String,Integer> dateMap = chartData.get(term);
        if (dateMap == null) {
          dateMap = new TreeMap<String, Integer>();
          chartData.put(term, dateMap);
        }

        String date = Bytes.toString(result.getRow());
        int freq = Integer.valueOf(Bytes.toString(entry.getValue()));
        dateMap.put(date, freq);
      }
    }

    // get all dates in the required range
    SortedSet<String> categories = TwitterControllerUtil.getAllDates("yearMonthDay", start, stop);

    // add zeros if required
    if (addZeros) {
      for (Map<String,Integer> dateMap : chartData.values()) {
        TwitterControllerUtil.addZeros(categories, dateMap);
      }
    }

    // put set and map into JSON
    JSONObject resultJson = new JSONObject();
    resultJson.put("categoriesLabel", "date");
    resultJson.put("categories", categories);
    resultJson.put("chartData", chartData);

    // write out the result JSON
    response.setContentType("text/plain; charset=UTF-8");
    OutputStream out = response.getOutputStream();
    out.write(Bytes.toBytes(jsonPrefix + resultJson.toString(JSON_INDENT_FACTOR) + jsonPostfix));
    out.flush();
    out.close();
  }

  /**
   * @param hbaseTablename
   */
  @Resource(name = "hbaseTablename")
  public void setHBaseTablename(String hbaseTablename) {
    this.hbaseTablename = hbaseTablename;
  }

  /**
   * @param hbaseColumnFamily
   */
  @Resource(name = "hbaseColumnFamily")
  public void setHBaseColumnFamily(String hbaseColumnFamily) {
    this.hbaseColumnFamily = hbaseColumnFamily.getBytes();
  }

  /**
   * @param hbaseColumn
   */
  @Resource(name = "hbaseColumn")
  public void setHBaseColumn(String hbaseColumn) {
    this.hbaseColumn = hbaseColumn.getBytes();
  }

  /**
   *
   * @param twitterstreamTablename
   */
  @Resource(name = "twitterstreamTablename")
  public void setTwitterstreamTablename(String twitterstreamTablename) {
    this.twitterstreamTablename = twitterstreamTablename;
  }

  /**
   * @param handler
   */
  @Autowired
  public void setHTablePoolHandler(HTablePoolHandler handler) {
    this.hTablePoolHandler = handler;
  }

  /**
   * @param exportService
   */
  @Autowired
  public void setExportService(TwitterVectorExporter exportService) {
    this.exportService = exportService;
  }
}
